package com.huawei.order;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * @author liucheng
 * @since 0.1 2023-11-12 20:49
 **/
public class OrderProducer {
    public static void main(String[] args) throws InterruptedException {
        String[] item = new String[]{"bread", "milk", "fruit", "clothes", "hat", "shoes"};

        Properties props = new Properties();
        props.put("serializer.class", ConstantUtils.SERIALIZER_CLASS_VALUE);
        props.put("bootstrap.servers", ConstantUtils.METADATA_BROKER_LIST);
        props.put("key.serializer", ConstantUtils.KEY_SERIALIZER);
        props.put("value.serializer", ConstantUtils.VALUE_SERIALIZER);
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        Order order = new Order();
        while (true) {
            int random = RandomUtils.getRandom(10);
            if (random == 0) {
                continue;
            }
            for (int i = 0; i < random; i++) {
                int orderRandom = RandomUtils.getRandom(random * 10);
                order.setName(item[RandomUtils.getRandom(6)]);
                order.setPrice(orderRandom);
            }

            // 向Kafka发送订单数据
            producer.send(new ProducerRecord<>(ConstantUtils.ORDER_TOPIC, order.getName() + " " + order.getPrice()));
            producer.flush();
            System.out.println(order);
            Thread.sleep(1000);
        }
    }
}
